package com.cryann.mqttserver.service.Impl;

import com.cryann.mqttserver.modle.Accessor;
import com.cryann.mqttserver.modle.AccessorStore;
import com.cryann.mqttserver.modle.MqttSendMessage;
import com.cryann.mqttserver.modle.Topic;
import com.cryann.mqttserver.repository.AccessorRepository;
import com.cryann.mqttserver.repository.TopicRepository;
import com.cryann.mqttserver.service.MqttAccessor;
import com.cryann.mqttserver.service.MsgStoreService;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.regex.Pattern;

/**
 * @author cRyann
 * @date 2018/12/20 0020 10:24
 * @description:
 */
@Slf4j
@Service("mqttAccessor")
public class MqttAccessorImpl implements MqttAccessor {

    @Autowired
    AccessorStore accessorStore;
    @Autowired
    AccessorRepository accessorRepository;
    @Autowired
    TopicRepository topicRepository;
    @Autowired
    MqttAccessor mqttAccessor;
    @Autowired
    MsgStoreService msgStoreService;

    @Override
    public boolean sendMsg(MqttSendMessage msg) {
        MqttPublishMessage message = (MqttPublishMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(msg.getMessageType(), msg.isDup(), msg.getQos(), msg.isRetain(), msg.getRemainingLength()),
                new MqttPublishVariableHeader(msg.getTopic(), msg.getMessageId()),
                Unpooled.buffer().writeBytes(msg.getByteBuf())
        );
        Set<Channel> channels = mqttAccessor.getAccessor(msg.getTopic());
        for (Channel channel : channels) {
            channel.writeAndFlush(message.retain());
        }
        return true;
    }

    @Override
    public boolean addAccessor(Channel channel, MqttConnectMessage msg) {
        String id = channel.id().asShortText();
        Accessor accessor = accessorRepository.findByChannelId(id);
        if ( accessor == null ) {
            accessor = new Accessor();
            accessor.setIdentifier(msg.payload().clientIdentifier());
            accessor.setChannelId(id);
            accessor.setAccessorName(msg.payload().userName());
            accessor.setAccessorIp(channel.remoteAddress().toString());
            accessor.setWillFlag(msg.variableHeader().isWillFlag());
            if ( msg.variableHeader().isWillFlag() ) {
                MqttMessage mqttMessage = MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(msg.variableHeader().willQos()), msg.variableHeader().isWillRetain(), 0),
                        new MqttPublishVariableHeader(msg.payload().willTopic(), msg.payload().hashCode()),
                        Unpooled.buffer().writeBytes(msg.payload().willMessageInBytes())
                );
                accessor.setWillMessage((MqttPublishMessage) mqttMessage);
            }
            accessor.setCleanSession(msg.variableHeader().isCleanSession());
            accessor.setKeepLive(msg.variableHeader().keepAliveTimeSeconds());
            accessor.setOnLine(true);
            accessor.setActiveTime(System.currentTimeMillis());
            accessor.setChannel(channel);
            accessorRepository.save(accessor);
            accessorStore.addContext(accessor);
            return true;
        }
        return false;
    }

    @Override
    public boolean delAccessor(Channel channel) {
        String id = channel.id().asShortText();
        Accessor accessor = accessorRepository.findByChannelId(id);
        if ( accessor != null && accessor.getCleanSession() ) {
            accessor.getTopicName().forEach(topic -> {
                topicRepository.findByTopicName(topic).ifPresent(t -> {
                    t.getAccessors().remove(accessor);
                    if ( t.getAccessors().size() < 1 ) {
                        topicRepository.delete(t);
                    } else {
                        topicRepository.saveAndFlush(t);
                    }
                });
            });
            accessorRepository.delete(accessor);
            accessorStore.delContext(accessor);
            return true;
        }
        return false;
    }

    @Override
    public Accessor getAccessor(Accessor accessor) {
        return null;
    }

    @Override
    public Accessor getAccessor(Channel channel) {
        return accessorRepository.findByChannelId(channel.id().asShortText());
    }

    @Override
    public Set<Channel> getAccessor(String topicFilter) {
        Set<Channel> channels = new HashSet<>();
        // MARK 根据Access 获取Channel
        getTopicNames(topicFilter).forEach(a -> {
            channels.add(accessorStore.getContext(a.getIdentifier()).getChannel());
        });
        return channels;
    }

    @Override
    public List<Accessor> getTopicNames(String topicFromMsg) {
        List<Accessor> result = new ArrayList<>();
        if ( StringUtils.containsAny(topicFromMsg, "+", "#") ) {
            try {
                throw new Exception("无效的主题名：" + topicFromMsg + ",因为它包含了无效的通配符");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        List<Topic> topicList = topicRepository.findAll();
        topicList.stream().forEach(t -> {
            if ( StringUtils.isNotBlank(t.getTopicName()) ) {
                String regex = StringUtils.replace(t.getTopicName(), "+", ".");
                regex = StringUtils.replace(regex, "#", ".*");
                boolean isMatch = Pattern.matches(regex, topicFromMsg);
                if ( isMatch ) {
                    result.addAll(t.getAccessors());
                }
            }
        });
        return result;
    }

    @Override
    public Accessor subTopic(Accessor accessor, String topic) {
        accessor.getTopicName().add(topic);
        accessorRepository.saveAndFlush(accessor);
        return accessor;
    }

    @Override
    public Accessor unSubTopic(Accessor accessor, List<String> topics) {
        Optional.ofNullable(accessor).ifPresent(a -> {
            accessor.getTopicName().removeAll(topics);
            accessorRepository.saveAndFlush(a);
        });
        topics.forEach(t -> {
            topicRepository.findByTopicName(t).ifPresent(topic -> {
                topic.getAccessors().remove(accessor);
                topicRepository.saveAndFlush(topic);
            });
        });
        return accessor;
    }

    @Override
    public Accessor getAccessorByUserName(String userName) {
        return null;
    }

    @Override
    public void updateAccessorByChannel(Channel channel) {
        accessorStore.getMap().values()
                .stream().filter(v -> v.getChannelId().equals(channel.id().asShortText()))
                .findFirst().ifPresent(accessor -> {
            accessor.setOnLine(true);
            accessor.setActiveTime(System.currentTimeMillis());
            accessorRepository.saveAndFlush(accessor);
            accessorStore.addContext(accessor);
        });
    }

    @Override
    public void willMessageSend(String identifier) {
        // TODO 获取 Accessor信息
        Optional.ofNullable(AccessorStore.instance().getContext(identifier)).ifPresent(accessor -> {
            if(accessor.getWillFlag()){
                Optional.of(accessor).ifPresent(a -> {
                    Set<Channel> channels = mqttAccessor.getAccessor(a.getWillMessage().variableHeader().topicName());
                    for (Channel channel : channels) {
                        channel.writeAndFlush(a.getWillMessage().retain());
                    }
                });
            }
        });
    }

}
